package com.ds.lens.flink.monitor;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.ds.lens.data.common.client.LensMonitorClient;
import com.ds.lens.data.common.client.LensMonitorClientWraper;
import com.ds.lens.data.common.constant.ChannelEnum;
import com.ds.lens.data.common.document.MailConfig;
import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.flink.monitor.eventsinkjob.function.notify.MonitorEventNotifier;
import com.ds.lens.flink.monitor.eventsinkjob.function.notify.mail.MailMonitorEventNotifier;
import com.ds.lens.flink.monitor.eventsinkjob.function.notify.onduty.OndutyClient;
import com.ds.lens.flink.monitor.eventsinkjob.function.notify.onduty.OndutyClientWraper;
import com.ds.lens.flink.monitor.eventsinkjob.function.notify.onduty.OndutyMonitorEventNotifier;
import com.ds.lens.flink.monitor.eventsinkjob.function.notify.sms.SMSMonitorEventNotifier;
import com.ds.lens.flink.monitor.eventsinkjob.function.notify.weixin.WeiXinConfig;
import com.ds.lens.flink.monitor.eventsinkjob.function.notify.weixin.WeiXinMonitorEventNotifier;
import com.ds.lens.flink.monitor.forecast.ForecastConfig;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.mail.EmailException;
import org.apache.http.HttpHost;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.util.StringUtils;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;

import static com.ds.lens.flink.monitor.MonitorJobConstants.*;

/**
 * @author ethan
 */
@Slf4j
public class MonitorJobContext {

    private static String CONFIG_XML_FILE = "classpath:lens-job-service.xml";

    private static String ELASTIC_JDBC_DATASOURCE = "lensElasticDataSource";

    private static String ELASTIC_JDBC_TEMPLATE = "lensElasticJdbcTemplate";

    private static String JSON_OBJECT = "jsonObjectMapper";

    private static String CONFIG_OBJECT = "jobConfigProperties";

    private static MonitorJobContext INSTANCE = new MonitorJobContext();

//    @Getter
//    private DruidDataSource elasticDruidDataSource;

//    @Getter
//    private JdbcTemplate elasticJdbcTemplate;

    @Getter
    private long checkpointInterval = 25000L;

    @Getter
    private long checkpointTimeout = 60000L;

    @Getter
    private int jobParallelism = 1;

    @Getter
    private int metricsJobParallelism = 1;

    @Getter
    private boolean enableCheckpoint = false;

    @Getter
    private Properties kafkaProperties;

    @Getter
    private ForecastConfig forecastConfig;

    @Getter
    private String monitorEventTopic;

    @Getter
    private String metricsTopic;

    @Getter
    private String externalEventTopic;

    @Getter
    private String monitorUrl;

    @Getter
    private LensMonitorClient monitorClient;

    @Getter
    private String agentId;

    @Getter
    private String serviceType;

    @Getter
    private String serviceKey;

    @Getter
    private OndutyClient ondutyClient;

    @Getter
    private String ondutyAuthorization;

    @Getter
    private int ticketExpiredInMinutes;

//    @Getter
//    private BinLogConfig binLogConfig;

    @Getter
    private MailConfig mailConfig;

    @Getter
    private WeiXinConfig weiXinConfig;

    @Getter
    private Map<Integer, MonitorEventNotifier<MonitorEvent, String>> channelMap = new HashMap<>();

    @Getter
    private int maxCacheSize;

    @Getter
    private int cacheExpiredInMinute = 1;

    @Getter
    private int maxBatchSize;

    @Getter
    private Gson gson;

    @Getter
    private Map<String, String> esConfig;

    @Getter
    private List<HttpHost> esTransportAddresses;

    @Getter
    private String esMonitorEventIndex;

    @Getter
    private String esMonitorEventType;

    @Getter
    private int lensDatasourceIntervalSecond;

//    @Getter
//    private ApmMetricClient apmMetricClient;

    @Getter
    private String esMetricdataIndex;

    @Getter
    private String esMetricdataType;

    @Getter
    private boolean printDebug;

    @Getter
    private String lensWebHost;


    private MonitorJobContext() {
        try {
            ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext(CONFIG_XML_FILE);
//            elasticDruidDataSource = applicationContext.getBean(ELASTIC_JDBC_DATASOURCE, DruidDataSource.class);
//            elasticJdbcTemplate = applicationContext.getBean(ELASTIC_JDBC_TEMPLATE, JdbcTemplate.class);

            Properties configProperties = applicationContext.getBean(CONFIG_OBJECT, Properties.class);
            initFlinkJobConfig(configProperties);
            initKafkaConfig(configProperties);
            initForecastConfig(configProperties);
//            initBinLogKafkaConfig(configProperties);
            initElasticSearch(configProperties);

            gson = new GsonBuilder().serializeNulls().create();
        } catch (Exception e) {
            log.error("job LensJobContext init failed.", e);
        }
    }

    public static MonitorJobContext getInstance() {
        return INSTANCE;
    }


    public static void shutdown() {
//        try {
//            if (INSTANCE.elasticDruidDataSource != null) {
//                INSTANCE.elasticDruidDataSource.close();
//                log.info("close elasticDruidDataSource succeeded.");
//            }
//        } catch (Exception e) {
//            log.error("close hikariDataSource failed", e);
//        }
    }

    private void initFlinkJobConfig(Properties configProperties) throws Exception {
        enableCheckpoint = Boolean.parseBoolean(configProperties.getProperty(
                MonitorJobConstants.JOB_CHECKPOINT_ENABLE_FIELD, "false"));
        jobParallelism = Integer.parseInt(configProperties.getProperty(
                MonitorJobConstants.JOB_DEFAULT_PARALLELISM_FIELD, Integer.toString(jobParallelism)));
        metricsJobParallelism = Integer.parseInt(configProperties.getProperty(
                MonitorJobConstants.JOB_METRICS_PARALLELISM_FIELD, Integer.toString(metricsJobParallelism)));

        checkpointInterval = Long.parseLong(configProperties.getProperty(
                MonitorJobConstants.JOB_CHECKPOINT_INTERVAL_FIELD, Long.toString(checkpointInterval)));
        checkpointTimeout = Long.parseLong(configProperties.getProperty(
                MonitorJobConstants.JOB_CHECKPOINT_TIMEOUT_FIELD, Long.toString(checkpointTimeout)));

        monitorUrl = configProperties.getProperty(MonitorJobConstants.MONITOR_DEFINE_SERVICE_URL, "");
        monitorClient = new LensMonitorClientWraper(monitorUrl);

//        String apmMetricUrl = configProperties
//                .getProperty(MonitorJobConstants.LENS_APM_METRIC_SERVICE_URL, "");
//        apmMetricClient = HttpServiceClient.getRestClient(ApmMetricClient.class, apmMetricUrl);

        agentId = configProperties.getProperty(MonitorJobConstants.ONDUTY_PARAM_AGENT_ID, "");
        serviceType = configProperties.getProperty(MonitorJobConstants.ONDUTY_PARAM_SERVICE_TYPE, "");
        serviceKey = configProperties.getProperty(MonitorJobConstants.ONDUTY_PARAM_SERVICE_KEY, "");
        String ondutyUrl = configProperties.getProperty(MonitorJobConstants.ONDUTY_SERVICE_URL, "");
        ondutyClient = new OndutyClientWraper(ondutyUrl);
        ondutyAuthorization = configProperties.getProperty(MonitorJobConstants.ONDUTY_AUTHORIZATION, "");
        ticketExpiredInMinutes = Integer.parseInt(configProperties.getProperty(MonitorJobConstants.ONDUTY_TICKET_EXPIRED_TIME, "30"));
        lensWebHost = configProperties.getProperty(LENS_WEB_HOST, "");
        channelMap.put(ChannelEnum.ON_DUTY.getValue(), new OndutyMonitorEventNotifier());
        initMailConfig(configProperties);
        initWeiXinConfig(configProperties);
        channelMap.put(ChannelEnum.SMS.getValue(), new SMSMonitorEventNotifier());
        maxCacheSize = Integer.parseInt(configProperties.getProperty(MonitorJobConstants.LENS_MAX_CACHE_SIZE, "1000"));
        cacheExpiredInMinute = Integer.parseInt(configProperties.getProperty(MonitorJobConstants.LENS_CACHE_EXPIRED_IN_MINUTE, "1"));
        maxBatchSize = Integer.parseInt(configProperties.getProperty(MonitorJobConstants.LENS_MAX_BATCH_SIZE, "500"));
        lensDatasourceIntervalSecond = Integer.parseInt(configProperties.getProperty(MonitorJobConstants.LENS_DATASOURCE_INTERVAL_SECOND, "10"));
        printDebug = Boolean.parseBoolean(configProperties.getProperty(MonitorJobConstants.LENS_PRINT_DEBUG_INFO, "false"));
    }

    private void initMailConfig(Properties configProperties) throws EmailException {
        mailConfig = new MailConfig();
        mailConfig.setMail_smtp_host(configProperties.getProperty(MonitorJobConstants.MAIL_SMTP_HOST, ""));
        mailConfig.setMail_smtp_port(Integer.parseInt(configProperties.getProperty(MonitorJobConstants.MAIL_SMTP_PORT, "0")));
        mailConfig.setMail_smtp_from(configProperties.getProperty(MonitorJobConstants.MAIL_SMTP_FROM, ""));
        mailConfig.setMail_smtp_username(configProperties.getProperty(MonitorJobConstants.MAIL_SMTP_USERNAME, ""));
        mailConfig.setMail_smtp_password(configProperties.getProperty(MonitorJobConstants.MAIL_SMTP_PASSWORD, ""));
        channelMap.put(ChannelEnum.MAIL.getValue(), new MailMonitorEventNotifier(mailConfig));
    }

    private void initWeiXinConfig(Properties configProperties) {
        if (Boolean.parseBoolean(configProperties.getProperty(MonitorJobConstants.WEIXIN_ENABLE, "false"))) {
            weiXinConfig = new WeiXinConfig();
            weiXinConfig.setUrl(configProperties.getProperty(MonitorJobConstants.WEIXIN_URL, "https://qyapi.weixin.qq.com/"));
            weiXinConfig.setCorpid(configProperties.getProperty(MonitorJobConstants.WEIXIN_CORPID, ""));
            weiXinConfig.setCorpsecret(configProperties.getProperty(MonitorJobConstants.WEIXIN_CORPSECRET, ""));
            weiXinConfig.setAgentid(Integer.parseInt(configProperties.getProperty(MonitorJobConstants.WEIXIN_AGENTID, "")));
            channelMap.put(ChannelEnum.WEIXIN.getValue(), new WeiXinMonitorEventNotifier(weiXinConfig));
        }
    }

    private void initKafkaConfig(Properties configProperties) {
        kafkaProperties = new Properties();
        kafkaProperties.setProperty(KAFKA_BROKER_SERVER_CONFIG_FIELD,
                configProperties.getProperty(MonitorJobConstants.JOB_KAFKA_SERVER_FIELD));
        kafkaProperties.setProperty(KAFKA_CONSUMER_GROUP_CONFIG_FIELD,
                configProperties.getProperty(MonitorJobConstants.JOB_KAFKA_GROUP_FIELD));
        monitorEventTopic = configProperties.getProperty(MonitorJobConstants.JOB_KAFKA_MONITOR_EVENT_TOPIC_FIELD);
        metricsTopic = configProperties.getProperty(MonitorJobConstants.JOB_KAFKA_METRICS_TOPIC_FIELD);
        externalEventTopic = configProperties.getProperty(MonitorJobConstants.JOB_KAFKA_EXTERNAL_EVENT_TOPIC_FIELD);
    }

    private void initForecastConfig(Properties configProperties) throws Exception {
        forecastConfig = new ForecastConfig();


        String esForecastDataType = configProperties.getProperty(ES_FORECAST_DATA_TYPE_CONFIG_FIELD);
        if (!StringUtils.hasText(esForecastDataType)) {
            throw new Exception("no forecast data type");
        }
        forecastConfig.setEsForecastDataType(esForecastDataType);
        String esForecastDataIndex = configProperties.getProperty(ES_FORECAST_DATA_INDEX_CONFIG_FIELD);
        if (!StringUtils.hasText(esForecastDataIndex)) {
            throw new Exception("no forecast data index");
        }
        forecastConfig.setEsForecastDataIndex(esForecastDataIndex);

        String esForecastKafkaDataIndex = configProperties.getProperty(ES_FORECAST_KAFKA_DATA_INDEX_CONFIG_FIELD);
        if (!StringUtils.hasText(esForecastKafkaDataIndex)) {
            throw new Exception("no forecast data index");
        }
        forecastConfig.setEsForecastKafkaDataIndex(esForecastKafkaDataIndex);
        String esForecastKafkaDataType = configProperties.getProperty(ES_FORECAST_KAFKA_DATA_TYPE_CONFIG_FIELD);
        if (!StringUtils.hasText(esForecastKafkaDataType)) {
            throw new Exception("no forecast data type");
        }
        forecastConfig.setEsForecastKafkaDataType(esForecastKafkaDataType);


        String esDynamicDetectionResultIndex = configProperties.getProperty(ES_DYNAMIC_DETECTION_RESULT_INDEX_CONFIG_FIELD);
        if (!StringUtils.hasText(esDynamicDetectionResultIndex)) {
            throw new Exception("no esDynamicDetectionResultIndex ");
        }
        forecastConfig.setEsDynamicDetectionResultIndex(esDynamicDetectionResultIndex);
        String esDynamicDetectionResultType = configProperties.getProperty(ES_DYNAMIC_DETECTION_RESULT_TYPE_CONFIG_FIELD);
        if (!StringUtils.hasText(esDynamicDetectionResultType)) {
            throw new Exception("no esDynamicDetectionResultType");
        }
        forecastConfig.setEsDynamicDetectionResultType(esDynamicDetectionResultType);


        forecastConfig.setForecastDataTopic(configProperties.getProperty(FORECAST_DATA_KAFKA_TOPIC_FIELD));
        forecastConfig.setForecastActualDataTopic(configProperties.getProperty(FORECAST_ACTUAL_DATA_KAFKA_TOPIC_FIELD));

        Properties properties = new Properties();
        properties.setProperty(KAFKA_BROKER_SERVER_CONFIG_FIELD,
                configProperties.getProperty(FORECAST_KAFKA_SERVER_FIELD));
        properties.setProperty(KAFKA_CONSUMER_GROUP_CONFIG_FIELD,
                configProperties.getProperty(FORECAST_KAFKA_GROUP_FIELD));
        properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        properties.put("sasl.jaas.config",
                "org.apache.kafka.common.security.plain.PlainLoginModule required " +
                        "username=\"" + configProperties.getProperty(FORECAST_KAFKA_USERNAME) + "\" " +
                        "password=\"" + configProperties.getProperty(FORECAST_KAFKA_PASSWORD) + "\";");
        forecastConfig.setKafkaProperties(properties);
    }
//    private void initBinLogKafkaConfig(Properties configProperties) {
//        binLogConfig = new BinLogConfig();
//        binLogConfig.setTopic(configProperties.getProperty(MonitorJobConstants.BINLOG_TOPIC));
//        binLogConfig.setAppID(Long.valueOf(configProperties.getProperty(MonitorJobConstants.BINLOG_APP_ID)));
//        binLogConfig.setAppKey(configProperties.getProperty(MonitorJobConstants.BINLOG_APP_KEY));
//        Properties properties = new Properties();
//        properties.setProperty(KAFKA_BROKER_SERVER_CONFIG_FIELD,
//                configProperties.getProperty(MonitorJobConstants.BINLOG_BOOTSTRAP));
//        properties.setProperty(KAFKA_CONSUMER_GROUP_CONFIG_FIELD,
//                configProperties.getProperty(MonitorJobConstants.BINLOG_GROUP_ID));
//        properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
//        properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
//        properties.put("sasl.jaas.config",
//                "org.apache.kafka.common.security.plain.PlainLoginModule required " +
//                        "username=\"" + configProperties.getProperty(MonitorJobConstants.BINLOG_USERNAME) + "\" " +
//                        "password=\"" + configProperties.getProperty(MonitorJobConstants.BINLOG_PASSWORD) + "\";");
//        binLogConfig.setKafkaProperties(properties);
//    }

    private void initElasticSearch(Properties configProperties) throws Exception {

        try {
            esMonitorEventIndex = configProperties.getProperty(ES_MONITOR_EVENT_INDEX_CONFIG_FIELD);
            if (!StringUtils.hasText(esMonitorEventIndex)) {
                throw new Exception("no monitorEvent index");
            }
            esMonitorEventType = configProperties.getProperty(ES_MONITOR_EVENT_TYPE_CONFIG_FIELD);
            if (!StringUtils.hasText(esMonitorEventType)) {
                throw new Exception("no monitorEvent type");
            }

            esMetricdataIndex = configProperties.getProperty(ES_METRIC_DATA_INDEX_CONFIG_FIELD);
            if (!StringUtils.hasText(esMetricdataIndex)) {
                throw new Exception("no metricdata index");
            }
            esMetricdataType = configProperties.getProperty(ES_METRIC_DATA_TYPE_CONFIG_FIELD);
            if (!StringUtils.hasText(esMetricdataType)) {
                throw new Exception("no metricdata type");
            }

            esConfig = new HashMap<>();
            esConfig.put(ES_CLUSTER_NAME_CONFIG_FIELD, configProperties.getProperty(
                    JOB_ES_CLUSTER_NODE_NAME_FIELD));
            esConfig.put(ES_BULK_SIZE_CONFIG_FIELD, configProperties.getProperty(
                    JOB_ES_BULK_SIZE_FIELD));
            esConfig.put(ES_INTERVAL_MS_CONFIG_FIELD, configProperties.getProperty(
                    JOB_ES_FLUSH_INTERVAL_FIELD));
            String esHttpNodes = configProperties.getProperty(JOB_ES_CLUSTER_HTTP_NODE_FIELD);
            if (!StringUtils.hasText(esHttpNodes)) {
                throw new Exception("document http node config invalid.");
            }
            String esSocketNodes = configProperties.getProperty(JOB_ES_CLUSTER_SOCKET_NODE_FIELD);
            if (!StringUtils.hasText(esSocketNodes)) {
                throw new Exception("document socket node config invalid.");
            }
            String[] esHttpNodeArray = esHttpNodes.split(COMMA_DELIMITER);
            if (esHttpNodeArray == null || esHttpNodeArray.length < 1) {
                throw new Exception("document node config was not found.");
            }
            String[] esSocketNodeArray = esSocketNodes.split(COMMA_DELIMITER);
            if (esSocketNodeArray == null || esSocketNodeArray.length < 1) {
                throw new Exception("document node config was not found.");
            }

            esTransportAddresses = new ArrayList<>();
            for (String esNode : esHttpNodeArray) {
                if (StringUtils.hasText(esNode)) {
                    String[] addressArray = esNode.split(SEMICOLON_DELIMITER);
                    if (addressArray == null || addressArray.length < 2) {
                        continue;
                    }
                    HttpHost address = new HttpHost(InetAddress.getByName(addressArray[0]), Integer.parseInt(addressArray[1]));
                    esTransportAddresses.add(address);
                }
            }
        } catch (UnknownHostException e) {
            throw new Exception("init elastic search failed.", e);
        }
    }
}
